package com.edu.yx.app.App_03_DwdApp.Dwd_03_Trade;

import com.edu.yx.app.APP_01_BaseApp.BaseSqlApp;
import com.edu.yx.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import static com.edu.yx.common.Constant.*;

public class Dwd_03_Trade_Payment_SucDetail extends BaseSqlApp {

    public static void main(String[] args) {
        new Dwd_03_Trade_Payment_SucDetail().init(
                3033,
                2,
                "Dwd_03_Trade_Payment_SucDetail"
        );
    }

    @Override
    protected void handle(StreamExecutionEnvironment env, StreamTableEnvironment tEnv) {
        //由于业务只是单纯的统计下单成功人数

        //1.拿ods
        readOdsDb(tEnv,"Dwd_03_Trade_Payment_SucDetail");

        //2.拿order_info 的user_id
        Table orderInfo = tEnv.sqlQuery("select " +
                "cast(data['id'] as string) order_id, " +
                "cast(data['user_id']as string) user_id," +
                " type, " +
                " ts " +
                " from ods_db " +
                " where `database` = 'edu' " +
                " and `table` = 'order_info' " +
                " and `type` = 'insert' "
        );
        tEnv.createTemporaryView("order_info_tmp",orderInfo);


        //2.拿支付信息表 由于计算的是支付人数，一个人一天可以成功支付多个，但是只能算一次
        Table paymentInfo = tEnv.sqlQuery("select " +
                "cast(data['id'] as string) payment_id, " +
                "cast(data['order_id']as string) order_id, " +
                "data['payment_type'] payment_type, " +
                "data['callback_time'] callback_time, " +
                " ts " +
                "from ods_db " +
                "where `database` = 'edu' " +
                "and `table` = 'payment_info' " +
                "and `type` = 'insert' " +
                "and `data`['payment_status'] = '1602' "
        );
        tEnv.createTemporaryView("payment_info_tmp",paymentInfo);

        //3.join
        Table result = tEnv.sqlQuery("select " +
                " oi.user_id," +
                " payment_id, " +
                " pi.order_id, " +
                " payment_type, " +
                " callback_time, " +
                " type," +
                " pi.ts " +
                " from payment_info_tmp pi " +
                " join order_info_tmp oi on oi.order_id = pi.order_id "
        );


        //4.发送到kafka
        tEnv.executeSql("create table dwd_trade_pay_detail_suc(" +
                "user_id string, " +
                "payment_id string, " +
                "order_id string, " +
                "payment_type string, " +
                "callback_time string, " +
                "type string, " +
                " ts bigint" +
                ")" + SQLUtil.getKafkaSink(TOPIC_DWD_TRADE_PAY_DETAIL_SUC)
        );

        result.executeInsert("dwd_trade_pay_detail_suc");

    }
}
